package com.ruoyi.business.task;

import com.alibaba.fastjson.JSON;
import com.ruoyi.business.constant.TradeConstants;
import com.ruoyi.business.domain.BusinessTradeRecord;
import com.ruoyi.business.dto.TicketDTO;
import com.ruoyi.business.enumType.PSIceOrderState;
import com.ruoyi.business.enumType.TriggerType;
import com.ruoyi.business.listener.AutoOrderSubscriber;
import com.ruoyi.business.listener.ProfitOrLossStopSubscriber;
import com.ruoyi.business.listener.TicketDataSaveSubscriber;
import com.ruoyi.business.service.IBusinessTradeConfigService;
import com.ruoyi.business.service.IBusinessTradeRecordService;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component("redisListenerTask")
public class RedisListenerTask {


    /**
     * 日志
     */
    private static final Logger logger = LoggerFactory.getLogger(RedisListenerTask.class);

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private AutoOrderSubscriber autoOrderSubscriber;

    @Autowired
    private ProfitOrLossStopSubscriber profitOrLossStopSubscriber;


    @Autowired
    private TicketDataSaveSubscriber ticketDataSaveSubscriber;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private IBusinessTradeConfigService businessTradeConfigService;

    @Autowired
    private IBusinessTradeRecordService businessTradeRecordService;


    /**
     * 从redis 消费消息
     * 1秒一次，保证消费顺序
     */
    @Scheduled(fixedDelay = 1000)
    public void popFromRedis() {
        while (true) {
            String message = redisTemplate.opsForList().rightPop(TradeConstants.TICKET_QUEUE);
            if (message != null) {
                saveRedisPrice(message);
                logger.info("从redis queue 取出消息:{}", message);
                autoOrderSubscriber.onTicket(message);
                profitOrLossStopSubscriber.onTicket(message);
                ticketDataSaveSubscriber.onTicket(message);
            }
        }
    }

    private void saveRedisPrice(String message) {
        List<TicketDTO> ticketDTOS = JSON.parseArray(message, TicketDTO.class);
        TicketDTO ticketDTO = ticketDTOS.get(0);
        BigDecimal price = new BigDecimal(String.valueOf(ticketDTO.getPrice()));
        BigDecimal open = new BigDecimal(String.valueOf(ticketDTO.getOpen()));
        Date marketTime = ticketDTO.getMarketTime();
        String keyOpen = TradeConstants.SYMBOL_PRICE_OPEN_KEY + DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD, marketTime);
        Object openObject = redisCache.getCacheObject(keyOpen);
        if (Objects.isNull(openObject)) {
            // 存入开盘价，兼容处理tick 和 bar 数据 open 价格不一致情况
            logger.info("存入行情日首次开盘价：{}", openObject);
            redisCache.setCacheObject(keyOpen, open, 6, TimeUnit.HOURS);
        }
        try {
            // 存入开仓 之后行情的最该高价和最低价 以及开盘价
            List<BusinessTradeRecord> businessTradeRecordList = businessTradeRecordService.selectTodayTradeRecord();
            // 开仓数据 部分成交或者已成交
            businessTradeRecordList = businessTradeRecordList.stream().filter(item -> TriggerType.AUTO_TAKE_POSITION_TRIGGER.equals(TriggerType.valueOf(item.getTriggerType()))).filter(item -> PSIceOrderState.PartFilled.equals(PSIceOrderState.valueOf(item.getDelegateState())) || PSIceOrderState.Filled.equals(PSIceOrderState.valueOf(item.getDelegateState()))).collect(Collectors.toList());
            if (!businessTradeRecordList.isEmpty()) {
                for (BusinessTradeRecord businessTradeRecord : businessTradeRecordList) {
                    String keyHigh = TradeConstants.SYMBOL_PRICE_HIGH_KEY + businessTradeRecord.getOrderNo();
                    String keyLow = TradeConstants.SYMBOL_PRICE_LOW_KEY + businessTradeRecord.getOrderNo();


                    Object highObject = redisCache.getCacheObject(keyHigh);
                    Object lowObject = redisCache.getCacheObject(keyLow);

                    if (Objects.nonNull(highObject)) {
                        BigDecimal highPrice = (BigDecimal) highObject;
                        logger.info(keyHigh + "获取目前最高价：{}", highPrice);
                        // 更新最高价
                        if (price.compareTo(highPrice) > 0) {
                            // 因为A 股 开市就4个小时 ，默认存储当前天行情数据
                            logger.info(keyHigh + "更新存入最高价：{}", price);
                            redisCache.setCacheObject(keyHigh, price, 6, TimeUnit.HOURS);
                        }
                    } else {
                        logger.info(keyHigh + "首次存入最高价：{}", price);
                        redisCache.setCacheObject(keyHigh, price, 6, TimeUnit.HOURS);
                    }

                    if (Objects.nonNull(lowObject)) {
                        BigDecimal lowPrice = (BigDecimal) lowObject;
                        logger.info(keyLow + "获取目前最低价：{}", lowPrice);
                        // 更新最低价
                        if (price.compareTo(lowPrice) < 0) {
                            // 因为A 股 开市就4个小时 ，默认存储当前天行情数据
                            logger.info(keyLow + "更新存入最低价：{}", price);
                            redisCache.setCacheObject(keyLow, price, 6, TimeUnit.HOURS);
                        }
                    } else {
                        logger.info(keyLow + "首次存入最低价：{}", price);
                        redisCache.setCacheObject(keyLow, price, 6, TimeUnit.HOURS);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("价格存入redis 报错", e);
        }
    }
}
